package hashmvp.superbloomfilter.pack;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;

import hashmvp.superbloomfilter.config.ConfigurationManager;



public class BloomFilter {
   //private BitSet bs;
   
   //private static int bitArraySize = 1000000000; //支持10亿数据 ，ram大小：10亿/8=1G字节,单个文件大小1G
   public static int numHashFunc = 2; //产生1个随机数 (依次是文件号，文件内的地址)
   
   public static List<Kindex> hashoutList=Collections.synchronizedList(new  ArrayList<Kindex>());
   public static List<Kindex> bufferList=new ArrayList<Kindex>();
   public static Kindex writekindexes = null;//new Kindex();
   public static Kindex readkindexes = null;//new Kindex();
   
   public static HashMap<Integer,List<KindexPos>>  posmap=new HashMap<Integer,List<KindexPos>>();
   
   //为contains(list)准备的pos map
   public static HashMap<Integer,List<KindexPos>>  kxposmap=new HashMap<Integer,List<KindexPos>>();
	
   //public static List<Kindex> hashinList=Collections.synchronizedList(new ArrayList<Kindex>());
   
   public static String bfParentFolder=null;//bf parent folder
   public static int bffilenums=0;//bf file number (文件个数10个）
   public static int bfzonenums=0;  //区块个数
   public static int bffilesize=0;  //bf file size(单个文件缺省尺寸是1G)
   public static int zonesize=0; //区块尺寸=filesize/zone num (一块区域至少支持1G,至多到10G,20G)
   public static boolean endflag=false; //线程结束标志 
   
   public static boolean writelockflag=false; //写锁标志
   
   public static final int MAX_FILENUM=10;
   public static final int MAX_ZONENUM=10;
   public static final long FILESIZE_10G=10*1024*1024*1024L; //10G文件
   public static final int FILESIZE_1G=1*1024*1024*1024; //1G文件
   public static final int FILESIZE_1M=1*1024*1024; //1M
		   
   public static final String DEAFULT_PARENTFOLDER="d:\\bfdata"; //bf data parent folder
   public static byte[] filezoneblock=null;
   //public static byte[] oncezoneblock=null;
   
   public static int oncereadsize=FILESIZE_1G;  //一次读1G      	
   public BloomFilter(){
	  ConfigurationManager.loadConfig("config.txt");
	  bfParentFolder=ConfigurationManager.getProperty("bf_parentpath")!=null?ConfigurationManager.getProperty("bf_parentpath"):DEAFULT_PARENTFOLDER;
	  String numstr=ConfigurationManager.getProperty("bf_filenums");
	  if (numstr!=null)
	     bffilenums = Integer.valueOf(numstr);
	  else
		 bffilenums = MAX_FILENUM;
	  
	  String zonenumstr=ConfigurationManager.getProperty("bf_zonenums");
	  if (zonenumstr!=null)
		  bfzonenums = Integer.valueOf(zonenumstr);
	  else
		  bfzonenums = MAX_ZONENUM;
	  
	  String sizestr=ConfigurationManager.getProperty("bf_filesize");
	  //size 是1G的倍数
	  if (sizestr!=null) {
		 if (sizestr.substring(sizestr.length()-1).equals("G")) {
		    sizestr= sizestr.substring(0,sizestr.length()-1);//去除"G"
		 }
		 bffilesize= Integer.valueOf(sizestr)*FILESIZE_1M;
	  }else {
		 bffilesize= FILESIZE_1M;
	  }
		  
	  zonesize=(int)((long)bffilesize /(long)bfzonenums);//used for writing file
	  
	  filezoneblock =new byte[bffilesize];//
	  
      //creat thread to write hashnum data to output file
      new Thread(new Runnable() {
    	  
    		@Override
    		public void run() {
    			// TODO Auto-generated method stub
    			while (!endflag){
    				if (writelockflag){
    					try {
							Thread.sleep(10);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
    					continue;
    				}
    				
    				if (hashoutList.size()==0)
    					continue;
    				
    				writelockflag=true;
    				//从hashout输出队列取出数据
    				bufferList.addAll(hashoutList);
    				hashoutList.clear();
    				writelockflag=false;
    				
    				//输出哈希数列表到文件
    				outputListtoFile(bufferList);
    				bufferList.clear();
    			}
    		}
    	}).start();
   }
 
   //产生初始空文件(耗时比较长)
   //读配置文件，按照配置文件生成空文件
   public void initworkzone(){
	   String dirpath = bfParentFolder;
	   File f1 = new File(dirpath);
	   if (!f1.exists()) {//目录不存在，创立目录
	      boolean ret=f1.mkdir();
	      if (!ret){
	         System.out.printf("making folder[%s] is not successful\n",dirpath);
	    	 return;
	      }
	   }
	   
	   for (int i=0;i<bffilenums;i++) {
		  String filepath= bfParentFolder+"\\"+(i+1); 
		  File f2 = new File(filepath);
		  if (!f2.exists()) {//文件不存在，创立缺省文件
			 try {
				f2.createNewFile();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				System.out.printf("making file[%s] is not successful\n",filepath);
				continue;
			}
		    makeDefaultFile(filepath);
		  }
	   }
	    	 
   }
   
   public void endwork(){
	   endflag=true;
   }
   
   public void add(String obj){
	  
	  if (obj==null)
		  return ;
	  
	  System.out.println(obj.toString());
	  //根据输入值得到若干组hash数
	  Kindex kindexes = getHashIndexes(obj,1);
	  Kindex kindexe2s = getHashIndexes(obj,1);
	  
	  if (kindexes==null)
		return ;
	  
	  while (writelockflag) {
		 try {
			Thread.sleep(10);//fix bug: jvm out of memory
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	  }
	  writelockflag=true;
	  kindexes.outputHashNums(); 
	  System.out.println("\n");
	  hashoutList.add(kindexes);
	  
	  writelockflag=false;
	  
	  
       // bs.set(index);
   }
 
   //支持单个数据项
   public boolean contains(String obj){
	 if (obj==null)
		return false;
	 
	 //System.out.println(obj.toString());
	 
	 Kindex kindexes = getHashIndexes(obj,0);
	 
	 if (kindexes==null)
		 return false;
	 
	 //kindexes.outputHashNums(); 
	  
	 if(isExistedHash(kindexes) == false)
         return false;
     
     return true;
   }
 
   //支持批量数据项
   public List<KxInfo> containlist(List<KxInfo> datalist){
	   //List<Kindex> kxlist=new ArrayList<Kindex>();
	   boolean flag=false;
	   for (int i=0;i<datalist.size();i++) {
		    KxInfo info=datalist.get(i);
		    Kindex ix=getHashIndexes(info.getItem(),0); 
		   //ix.setIndex(info.getIndex());
			 
			 //kindexes.outputHashNums(); 
		    long sttime=System.currentTimeMillis();
	          
			flag=isExistedHash(ix);
		    info.setResult(flag);
		    datalist.set(i, info);	    
		    long endtime=System.currentTimeMillis();
	        System.out.println("take"+(endtime-sttime)+"ms");
	        
       }
	   
	    
	   //parsekxList(kxlist,datalist);
	   return datalist;
   }
   
   public void parsekxList(List<Kindex> kxlist,List<KxInfo> datalist){
	   for (Kindex kitem:kxlist){
			  
			  int fileno= Math.abs((int)(kitem.getValue(0) % bffilenums))+1;
			  //int zoneno= Math.abs((int)(kitem.getValue(1) % bfzonenums));
			  int pos=Math.abs((int)(kitem.getValue(1) ));//bitpos
			  //long endpos=Math.abs((long)(kitem.getValue(3) % (long)zonesize));
			  KindexPos kpos=new KindexPos();
			  kpos.setFileno(fileno);
			  kpos.setPos(pos);
			  kpos.setIndex(kitem.getIndex());
			  insertoPosMap(kxposmap,kpos);
			   
	  }
	   
	  Iterator iter = kxposmap.entrySet().iterator();
	  while (iter.hasNext()) {
	      Map.Entry entry = (Map.Entry) iter.next();
	      
	      //int fileno = (int)entry.getKey();
	      
	      //得到属于同一个文件的list
	      List<KindexPos> poslist = (List<KindexPos>)entry.getValue();
	      
	      //按区块号排序
	      long sttime=System.currentTimeMillis();	        
	      poslist= sortPoslist(poslist);
	      if (poslist.size()==0)
	    	  continue;
	      
	      long endtime=System.currentTimeMillis();
	      System.out.println("sort take"+(endtime-sttime)+"ms");
	      
	      int fileno=poslist.get(0).fileno;
	      
	      //add writefile lock
	      //Existinfile(fileno,poslist);//read data to memory
	    	  
	    	  
	    
	      
	  }
	  
	  System.out.println("reading data end");
      return ;
   }
   
   /*采用MD5散列作为java随机数生成器的种子并取k个随机数作为索引*/
   //flag:0--read,1--write
   public  Kindex getHashIndexes(String obj,int flag){
     long seed = 0;
     byte[] digest;
     try {
           MessageDigest md = MessageDigest.getInstance("MD5");
        
           byte[] objbytes=obj.toString().getBytes();
        
           Kindex kindexes = null;//new Kindex();
           if (flag==1) {
        	   writekindexes=new Kindex();
        	   kindexes=writekindexes;
           }
           else {
        	   readkindexes=new Kindex();
        	   kindexes=readkindexes;
           }
           
           md.update(obj.toString().getBytes());
           digest = md.digest(); //Completes the hash computation 
           long indexv =0;           
           for(int i=0;i<numHashFunc;i++) {
              seed = seed^(((long)(digest[i] & 0xFF)) << (8*i));
           }
           
           Random gen = new Random(seed);
           for (int i=0;i<numHashFunc;i++) {
              indexv = gen.nextLong();
              kindexes.setValue(indexv, i);
              
           }
           
           
           return kindexes;
     } catch (NoSuchAlgorithmException e) {
         System.out.println("no md5");
    	 e.printStackTrace();
     }
     
     return null;
   }
 
  
   //读测试文件
   public Map<Integer,String> readFile(String filePath){
        
	 BufferedReader br;
     Map<Integer,String> map = new HashMap<Integer,String>();
     try {
         br = new BufferedReader(new InputStreamReader(
           new FileInputStream(filePath)));
         int i = 0;
         for (String line = br.readLine(); line != null; line = br.readLine()) {
             map.put(i++, line);
         }
         br.close();
     } catch (FileNotFoundException e) {
        e.printStackTrace();
     } catch (IOException e) {
        e.printStackTrace();
     }
        return map;
   }
 
   public static void main(String[] args) {
     
	 BloomFilter bf = new BloomFilter();
	 bf.initworkzone();
     
     
     Map<Integer,String> map = bf.readFile(".\\bfexample.txt");
     for(Map.Entry<Integer, String> m : map.entrySet())
        bf.add(m.getValue());
    
     bf.add(null);
     bf.add("");
     bf.add("b");
     
     try {
		Thread.sleep(200); //初次加数据平均20至30毫秒延迟，数据越多，需要越多延迟。
	 } catch (InterruptedException e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	 }
     
    
     boolean flag = bf.contains("test is 69");
     System.out.println(flag);
     
     flag = bf.contains("test is 699");
     System.out.println(flag);
     
     flag = bf.contains("g");
     System.out.println(flag);
     
     flag = bf.contains("this");
     System.out.println(flag);
     
     flag = bf.contains("t56");
     System.out.println(flag);
     
     
     flag = bf.contains("this is third superbloomfilters long long long filter");
     System.out.println(flag);
     flag = bf.contains("test index56");
     System.out.println(flag);     
     
     flag = bf.contains("a super string contans number and char. and have many signals like ^^&&**%$###@@");
     System.out.println(flag);     
     
     flag = bf.contains("a super string 43434 ag contans number and char. and have many signals like ^^&&**%$###@@!@^&&");
     System.out.println(flag);     
     
     bf.endwork();
     
  }
   
  public  void outputListtoFile(List<Kindex> indexlist){
	  
	  //HashMap<Integer,List<KindexPos>>  posmap=new HashMap<Integer,List<KindexPos>>();
	   
	  for (Kindex kitem:indexlist){
		  
		  int fileno= Math.abs((int)(kitem.getValue(0) % bffilenums))+1;
		  int pos=(int)Math.abs((long)(kitem.getValue(1) %(long) bffilesize));//bitpos
		  KindexPos kpos=new KindexPos();
		  kpos.setFileno(fileno);
		  kpos.setPos(pos);
		  insertoPosMap(posmap,kpos);
		   
	  }
	  
	  Iterator iter = posmap.entrySet().iterator();
	  while (iter.hasNext()) {
	      Map.Entry entry = (Map.Entry) iter.next();
	      int fileno = (int)entry.getKey();
	      
	      List<KindexPos> poslist = (List<KindexPos>)entry.getValue();
	      
	      String filepath = bfParentFolder+"\\"+String.valueOf(fileno);
	      File f = new File(filepath);
	         
	      if (!f.exists()){
	        	System.out.printf("file[%s] is not existed,need creating\n",filepath);
				continue;
	      }
	      
	      RandomAccessFile bf = null;  
	      try{
	        	 bf = new RandomAccessFile(filepath,"rw"); 
	        	 for (KindexPos kitem:poslist) {
	        	    writeHashnumstoFile(bf,kitem.getFileno(),kitem.getPos());
	        	 }
	        	 bf.close();
	      } 
	      catch (FileNotFoundException e){
	        	 System.out.printf("file[%s] is not found\n",filepath);
	        	 continue;
	      }catch (IOException e) {
					// TODO Auto-generated catch block
				 System.out.printf("file[%s] op is not successful\n",filepath);
				 continue;
		  }	
	              	
	        
	    	
	  }
	  posmap.clear();
	 
  }
  
  public List<KindexPos> sortPoslist (List<KindexPos> poslist){
		Collections.sort(poslist, new Comparator<KindexPos>() {
	           //sort asc
		       public int compare(KindexPos h1, KindexPos h2) {
		           if (h1.pos>h2.pos)
		        	   return 1;
		           else if (h1.pos==h2.pos)
		        	   return 0;
		           else 
		        	   return -1;
		       }
		   });
		
		return poslist;
		
	
  }
	
  public void insertoPosMap(HashMap<Integer,List<KindexPos>> posmap,KindexPos kpos){
	  
	  List<KindexPos> samefilelist=null; //同属一个文件的pos list
	  int key=kpos.getFileno();
	  if (posmap.get(key)!=null) {
		  samefilelist= posmap.get(key);
	  }else {
		  samefilelist =new ArrayList<KindexPos>();
	  }
	  samefilelist.add(kpos);
	  posmap.put(key, samefilelist);
	  
  }
  
  public boolean writeHashnumstoFile(RandomAccessFile bf,int fileno,long bitpos) throws IOException{
	  
 	    int stbytepos= (int)(bitpos / 8);
	    //bf.seek(zoneno*FILESIZE_1G+stbytepos);
	    int pos=(int)stbytepos;
	    bf.seek(pos);
	
	    int sb=bf.readByte();//文件指针skip 1
	    int offset=(int)bitpos % 8;
	    sb=sb|(1<<offset);
	    bf.seek(pos);   //需要重新定位
    	bf.writeByte(sb);
    	
	    return true;
  }
  
  public boolean isExistedHash(Kindex kitem) {
	  int fileno= Math.abs((int)(kitem.getValue(0) % bffilenums))+1;
	  //int zoneno= Math.abs((int)(kitem.getValue(1) % bfzonenums));
	  long stbitpos=Math.abs((long)(kitem.getValue(1) %(long) bffilesize));//bitpos
	  //long endpos=Math.abs((long)(kitem.getValue(3) % (long)zonesize));
	  return isExistedHashnum(fileno,stbitpos);
  }
  
  public boolean isExistedinBlock(long stbitpos,long endbitpos){
	    
	    int stbytepos= (int)(stbitpos / 8);
	   	
	    int pos=stbytepos;
	   	
	   	byte sb=filezoneblock[pos];
	   	int offset=(int)stbitpos % 8;
	   	sb=(byte)(sb&(1<<offset));
	   	if (sb==0) {
	   	   return false;
	   	}
	   	
	   	int endbytepos=(int)(endbitpos / 8);
	   	pos=endbytepos;
	   	
	   	byte eb=filezoneblock[pos];
	   	offset=(int)endbitpos % 8;            	
	   	eb=(byte)(eb&(1<<offset));	
	   	if (eb==0) {
		   return false;
		}
	   	
	   	return true;
		   	
  }
  
  public boolean isExistedHashnum(int fileno,long bitpos){
	  
	 String filepath = bfParentFolder+"\\"+String.valueOf(fileno);
 	 File f = new File(filepath);
      
      if (!f.exists()){
     	 System.out.printf("file[%s] is not existed,need creating\n",filepath);
	     return false;
				 
      }
 	 
      
      RandomAccessFile bf = null;  
      try{
    	    long sttime=System.currentTimeMillis();
       	
    	    bf = new RandomAccessFile(filepath,"r"); 
         	
         	
         	long stbytepos= (long)(bitpos / 8);
         	//bf.seek(zoneno*FILESIZE_1G+stbytepos);
         	long pos=stbytepos;
         	bf.seek(pos);
         	
         	byte sb=bf.readByte();
         	
         	int offset=(int)(bitpos % 8);
         	sb=(byte)(sb&(1<<offset));
         	if (sb==0) {
         	   bf.close();
         	   return false;
         	}
         	
         	bf.close();
            
      }catch (FileNotFoundException e){
     	  System.out.printf("file[%s] is not found\n",filepath);
     	  return false;
      }catch (IOException e) {
				// TODO Auto-generated catch block
		   System.out.printf("file[%s] op is not successful\n",filepath);
		   return false;
		 }	
      
      return true;
  }

  

  public boolean makeDefaultFile(String filepath) {
	  RandomAccessFile bf = null;  
	  
      try{
     	    bf = new RandomAccessFile(filepath,"rw");
     	    byte[] bs=new byte[FILESIZE_1M]; //1G:文件最小单位
     	    int num=(int)(bffilesize/FILESIZE_1M);
     	    for (int i=0;i<num;i++) { //10G:循环10次
     	      bf.write(bs,0,FILESIZE_1M);
     	    }
     	    bf.close();
      }catch (FileNotFoundException e){
     	 System.out.printf("file[%s] is not found\n",filepath);
     	 return false;
      }catch (IOException e) {
				// TODO Auto-generated catch block
	     System.out.printf("file[%s] op is not successful\n",filepath);
		 return false;
	  }	
      
      return true;
  }
    
  static class  Kindex {
		 private int numHashFunc = 2; //产生2个随机数
		 private long[] hashnums;
		 private int index;
		
		 public Kindex() {
			 hashnums =new long[numHashFunc];
		 }
		
		 public void setIndex(int ix) {
			  index=ix;
		 }
		  
		 public int getIndex() {
			  return index;
		 }
		
		 public void setValue(long value,int index) {
			 hashnums[index]=value;
		 }
		 
		 public long getValue(int index) {
			 return hashnums[index];
		 }
		
		 public void outputHashNums() {
			 for (int i=0;i<numHashFunc;i++) {
			    System.out.print(hashnums[i]+" ");
			 }
			 System.out.println("");
		 }
  }
  
  static class  KindexPos {
        private int  fileno;
        //private int  zoneno;
        private int pos;
        //private long endpos;
        private int index;
  	  
        public void setFileno(int no) {
        	fileno=no;
        }
        
        public int getFileno() {
        	return this.fileno;
        }
        
        
        
        public void setPos(int pos) {
        	this.pos=pos;
        }
        
        public int getPos() {
        	return this.pos;
        }
        
        
        public void setIndex(int ix) {
  		  index=ix;
  	    }
  	  
  	    public int getIndex() {
  		  return index;
  	    }
     
  }
  
  
  public boolean isExistedHashnum2(byte[] buffer,int bitpos){
	  
		   
	         	
          int stbytepos= (int)(bitpos / 8);
          byte sb=buffer[stbytepos];
          int offset=(int)bitpos % 8;
          sb=(byte)(sb&(1<<offset));
          if (sb==0) {
         	   //bf.close();
         	   return false;
          }
         	
	         	
	      return true;
   }


  
}

 

 


